package com.dy.spring.boot.configration;



import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 描述 com.bwton.ttsp.pay.front.lisener
 *
 * @author DengYang
 * @date 2018/4/13 15:59
 * @since JDK1.8.0_161
 */
@Slf4j
public class MessageLisener implements MessageListenerConcurrently {

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        // 如果没有return success ，consumer会重新消费该消息，直到return success
        String body = "";
        try {
            Message msg = msgs.get(0);
            body = new String(msg.getBody(), "utf-8");
            log.info("接收到消息 --> Topic = {},Tags={},Key={},data={}", msg.getTopic(), msg.getKeys(), msg.getTags(), body);
//            if (MQConstant.TopicEnum.PAY_TOPIC.getValue().compareTo(msg.getTopic()) == 0) {
//                ConsumeRequestDto model = gson.fromJson(body, ConsumeRequestDto.class);
//                String msgConsume = payOrderEventService.consume(model);
//                if (TransConstant.MSG_CONSUME_SUCESS.compareTo(msgConsume) == 0) {
//                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//                } else {
//                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//                }
//            }
//            if (MQConstant.TopicEnum.NOTIFY_PAY_TOPIC.getValue().compareTo(msg.getTopic()) == 0) {
//                NotifyRequestModel model = gson.fromJson(body, NotifyRequestModel.class);
//                String msgConsume = payOrderEventService.consumeNotify(model);
//                if (TransConstant.MSG_CONSUME_SUCESS.compareTo(msgConsume) == 0) {
//                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//                } else {
//                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//                }
//            }

            log.info("接收到消息处理成功 --> Topic = {},Tags={},Key={},data={}", msg.getTopic(), msg.getKeys(), msg.getTags(), body);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            log.error("pay-event中MessageLisener消费交易消费失败{},异常{}", body, e.getMessage());
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}
